package com.fsxgt.datax;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.datax.common.constant.PluginType;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.TaskPluginCollector;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.core.statistics.communication.Communication;
import com.alibaba.datax.core.statistics.plugin.task.AbstractTaskPluginCollector;
import com.alibaba.datax.core.taskgroup.runner.AbstractRunner;
import com.alibaba.datax.core.taskgroup.runner.ReaderRunner;
import com.alibaba.datax.core.taskgroup.runner.WriterRunner;
import com.alibaba.datax.core.transport.channel.Channel;
import com.alibaba.datax.core.transport.transformer.TransformerExecution;
import com.alibaba.datax.core.util.ClassUtil;
import com.alibaba.datax.core.util.ErrorRecordChecker;
import com.alibaba.datax.core.util.FrameworkErrorCode;
import com.alibaba.datax.core.util.container.CoreConstant;
import com.alibaba.datax.dataxservice.face.domain.enums.State;
import com.alibaba.datax.plugin.rdbms.reader.Constant;
import com.alibaba.datax.plugin.rdbms.reader.util.ReaderSplitUtil;
import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
import com.alibaba.datax.plugin.rdbms.writer.Key;
import com.alibaba.datax.plugin.rdbms.writer.util.WriterUtil;

public class TaskExecutor {

	private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class);

	private Configuration taskConfig;
	


	private Channel channel;

	private Thread readerThread;

	private Thread writerThread;

	private Thread monitorThread;

	private ReaderRunner readerRunner;

	private WriterRunner writerRunner;

	private static AtomicInteger index = new AtomicInteger();

	/**
	 * 该处的taskCommunication在多处用到： 1. channel 2. readerRunner和writerRunner 3.
	 * reader和writer的taskPluginCollector
	 */
	private Communication taskCommunication;

	public Communication getCommunication() {
		return taskCommunication;
	}

	private String channelClazz;

	private Configuration configuration;

	private String taskCollectorClass;

	private ErrorRecordChecker errorRecordChecker;
	
	public void doStop() {
		
		readerThread.interrupt();

	}

	public TaskExecutor(Configuration configuration) {
		// 获取该taskExecutor的配置
		


		this.configuration = configuration;

		this.taskConfig = configuration.getListConfiguration(CoreConstant.DATAX_JOB_CONTENT).get(0);

		this.errorRecordChecker = new ErrorRecordChecker(configuration);



		this.taskCommunication = new Communication();
		this.taskCollectorClass = this.configuration
				.getString(CoreConstant.DATAX_CORE_STATISTICS_COLLECTOR_PLUGIN_TASKCLASS);
		this.channelClazz = this.configuration.getString(CoreConstant.DATAX_CORE_TRANSPORT_CHANNEL_CLASS);

		this.channel = ClassUtil.instantiate(channelClazz, Channel.class, configuration);
		this.channel.setCommunication(this.taskCommunication);

		GlobalConfig.init(configuration);

		/**
		 * 获取transformer的参数
		 */

//        List<TransformerExecution> transformerInfoExecs = TransformerUtil.buildTransformerInfo(taskConfig);

		List<TransformerExecution> transformerInfoExecs = new ArrayList();

		this.monitorThread = new Thread(new Runnable() {
			@Override
			public void run() {
				// TODO Auto-generated method stub
				try {
					while (!taskCommunication.isFinished()) {
						errorRecordChecker.checkRecordLimit(taskCommunication);
						errorRecordChecker.checkPercentageLimit(taskCommunication);
						Thread.sleep(500);
					}
				} catch (Exception e) {
					readerRunner.shutdown();
				}
			}

		}, String.format("dg-datax-monitor-%d", index.getAndIncrement()));

		/**
		 * 生成writerThread
		 */
		writerRunner = (WriterRunner) generateRunner(PluginType.WRITER);

		this.writerThread = new Thread(writerRunner, String.format("dg-datax-writer-%d", index.getAndIncrement()));


		// 通过设置thread的contextClassLoader，即可实现同步和主程序不通的加载器
//        this.writerThread.setContextClassLoader(LoadUtil.getJarLoader(
//                PluginType.WRITER, this.taskConfig.getString(
//                        CoreConstant.JOB_WRITER_NAME)));
		/**
		 * 生成readerThread
		 */
		readerRunner = (ReaderRunner) generateRunner(PluginType.READER, new ArrayList());
		this.readerThread = new Thread(readerRunner, String.format("dg-datax-reader-%d", index.getAndIncrement()));


		/**
		 * 通过设置thread的contextClassLoader，即可实现同步和主程序不通的加载器
		 */
//        this.readerThread.setContextClassLoader(LoadUtil.getJarLoader(
//                PluginType.READER, this.taskConfig.getString(
//                        CoreConstant.JOB_READER_NAME)));
	}

	public void doStart() throws InterruptedException {

		this.monitorThread.start();

		this.writerThread.start();

//        // reader没有起来，writer不可能结束
//        if (!this.writerThread.isAlive() || this.taskCommunication.getState() == State.FAILED) {
//            throw DataXException.asDataXException(
//                    FrameworkErrorCode.RUNTIME_ERROR,
//                    this.taskCommunication.getThrowable());
//        }

		this.readerThread.start();

//        // 这里reader可能很快结束
//        if (!this.readerThread.isAlive() && this.taskCommunication.getState() == State.FAILED) {
//            // 这里有可能出现Reader线上启动即挂情况 对于这类情况 需要立刻抛出异常
//            throw DataXException.asDataXException(
//                    FrameworkErrorCode.RUNTIME_ERROR,
//                    this.taskCommunication.getThrowable());
//        }

		this.writerThread.join();
	
		if(!taskCommunication.getState().equals(State.SUCCEEDED)) {
			throw DataXException.asDataXException(
                    FrameworkErrorCode.PLUGIN_RUNTIME_ERROR,
                    "");
		}

		errorRecordChecker.checkRecordLimit(taskCommunication);
		errorRecordChecker.checkPercentageLimit(taskCommunication);
		


	}

	private AbstractRunner generateRunner(PluginType pluginType) {
		return generateRunner(pluginType, null);
	}

	private AbstractRunner generateRunner(PluginType pluginType, List<TransformerExecution> transformerInfoExecs) {
		AbstractRunner newRunner = null;
		TaskPluginCollector pluginCollector;

		switch (pluginType) {
		case READER:
			newRunner = LoadUtil.loadPluginRunner(pluginType, this.taskConfig.getString(CoreConstant.JOB_READER_NAME));

			Configuration config = this.taskConfig.getConfiguration(CoreConstant.JOB_READER_PARAMETER);
			DataBaseType dataBaseType = LoadUtil.getDataBaseType(pluginType,
					this.taskConfig.getString(CoreConstant.JOB_READER_NAME));
			ReaderOriginalConfPretreatmentUtil.doPretreatment(config, dataBaseType);

			newRunner.setJobConf(ReaderSplitUtil.doSplit(config, 1).get(0));

			newRunner.getJobConf().set(Constant.FETCH_SIZE, config.getInt(Constant.FETCH_SIZE, 3000));

			pluginCollector = ClassUtil.instantiate(taskCollectorClass, AbstractTaskPluginCollector.class,
					configuration, this.taskCommunication, PluginType.READER);




			((FastReaderRunner) newRunner).setRecordSender(new FastBufferedRecordExchanger(this.channel, pluginCollector));

			/**
			 * 设置taskPlugin的collector，用来处理脏数据和job/task通信
			 */
			newRunner.setTaskPluginCollector(pluginCollector);
			break;
		case WRITER:
			newRunner = LoadUtil.loadPluginRunner(pluginType, this.taskConfig.getString(CoreConstant.JOB_WRITER_NAME));

			config = this.taskConfig.getConfiguration(CoreConstant.JOB_WRITER_PARAMETER);
			dataBaseType = LoadUtil.getDataBaseType(pluginType,
					this.taskConfig.getString(CoreConstant.JOB_WRITER_NAME));
			WriterOriginalConfPretreatmentUtil.doPretreatment(config, dataBaseType);

			newRunner.setJobConf(WriterUtil.doSplit(config, 1).get(0));

			pluginCollector = ClassUtil.instantiate(taskCollectorClass, AbstractTaskPluginCollector.class,
					configuration, this.taskCommunication, PluginType.WRITER);
			
			((FastWriterRunner) newRunner)
					.setRecordReceiver(new FastBufferedRecordExchanger(this.channel, pluginCollector));
			/**
			 * 设置taskPlugin的collector，用来处理脏数据和job/task通信
			 */
			newRunner.setTaskPluginCollector(pluginCollector);
			break;
		default:
			throw DataXException.asDataXException(FrameworkErrorCode.ARGUMENT_ERROR,
					"Cant generateRunner for:" + pluginType);
		}


		newRunner.setRunnerCommunication(this.taskCommunication);

		return newRunner;
	}

}